package etcd

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"

	"github.com/lukeaxu/discover/registry"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/client/v3/concurrency"
)

type RegistryOpt func(*Registry)

type Registry struct {
	client  *clientv3.Client
	sess    *concurrency.Session
	ttl     int // 从服务器获取一个租约的最大延迟
	cancels map[string]func()
	mutex   sync.Mutex
}

func WithLeaseTTL(ttl int) RegistryOpt {
	return func(r *Registry) {
		r.ttl = ttl
	}
}

func NewRegistry(client *clientv3.Client, opts ...RegistryOpt) (registry.Registry, error) {
	r := &Registry{
		client:  client,
		ttl:     60,
		cancels: map[string]func(){},
	}
	for _, opt := range opts {
		opt(r)
	}
	// Session 表示一个由客户端维持的租约，在客户端的整个生命周期中保持活动状态。
	sess, err := concurrency.NewSession(client, concurrency.WithTTL(r.ttl))
	if err != nil {
		return nil, err
	}
	r.sess = sess
	return r, nil
}

func (r *Registry) Register(ctx context.Context, si registry.ServiceInstance) error {
	val, err := json.Marshal(si)
	if err != nil {
		return err
	}
	_, err = r.client.Put(ctx, r.keyForServiceInstance(si), string(val), clientv3.WithLease(r.sess.Lease()))
	return err
}

func (r *Registry) UnRegister(ctx context.Context, si registry.ServiceInstance) error {
	_, err := r.client.Delete(ctx, r.keyForServiceInstance(si))
	return err
}

func (r *Registry) Subscribe(ctx context.Context, sn string) (<-chan registry.Event, error) {
	// 如何将服务名字变成在注册中心上的 key，这是谁的职责？
	// 对于用户来说，只知道一个服务名，具体如何保存在注册中心，是以 kv 的形式保存在注册中心还是以其他方式保存
	// 完全由不同的注册中心来决定，对于 etcd 来说就是要把服务名字变成 key,然后通过 key 获取注册的信息
	// Mark: 因为同一个服务可能由挺多实例注册，注册的名字要做一下区分（前缀一样）
	// 未了拿到服务的多个实例，所以按照前缀匹配从注册中心取数据

	// 因为基于 etcd 的订阅过程一直发生，知道关闭注册中心，所以不用传入的 ctx
	ctx, cancel := context.WithCancel(ctx)
	// 因为可能会订阅多个服务，关闭的时候可能关闭多个 watch 的协程，cancels可能被同时操作，所以要加锁
	r.mutex.Lock()
	r.cancels[r.keyForServiceName(sn)] = cancel
	r.mutex.Unlock()
	ctx = clientv3.WithRequireLeader(ctx)
	watchChann := r.client.Watch(ctx, r.keyForServiceName(sn), clientv3.WithPrefix())
	// Mark: 需要不需要设置 buffer size?
	// 这个 Channle 需要在 Resovler中处理，为了不让 Resolver 依赖具体的 registry 的实现，所以不能直接返回上面的watchChann
	eventChann := make(chan registry.Event)

	go func() {
		for {
			select {
			case resp := <-watchChann:
				if resp.Err() != nil {
					return
				}
				if resp.Canceled {
					return
				}
				for range resp.Events {
					eventChann <- registry.Event{}
				}
			case <-ctx.Done():
				r.UnSubscribe(context.Background(), r.keyForServiceName(sn))
				return
			}
		}
	}()
	return eventChann, nil
}

func (r *Registry) UnSubscribe(ctx context.Context, sn string) error {
	key := r.keyForServiceName(sn)
	r.mutex.Lock()
	cancel, ok := r.cancels[key]
	if ok {
		delete(r.cancels, key)
	}
	r.mutex.Unlock()
	cancel()
	return nil
}

func (r *Registry) Close() error {
	r.mutex.Lock()
	cancels := r.cancels
	r.cancels = nil
	r.mutex.Unlock()
	for _, cancelfunc := range cancels {
		cancelfunc()
	}
	return r.sess.Close()
}

func (r *Registry) ListServices(ctx context.Context, sn string) ([]registry.ServiceInstance, error) {
	resp, err := r.client.Get(ctx, r.keyForServiceName(sn), clientv3.WithPrefix())
	if err != nil {
		return nil, err
	}
	ret := make([]registry.ServiceInstance, len(resp.Kvs))
	for i, kv := range resp.Kvs {
		err := json.Unmarshal(kv.Value, &ret[i])
		if err != nil {
			return nil, err
		}
	}
	return ret, nil
}

// keyForServiceName 获取服务的时候，据此从etcd找相关服务
func (r *Registry) keyForServiceName(sn string) string {
	return fmt.Sprintf("%s/", sn)
}

// keyForServiceInstance 注册服务的时候的 key
func (r *Registry) keyForServiceInstance(si registry.ServiceInstance) string {
	return fmt.Sprintf("%s/%s", si.Name, si.Addr)
}
